package com.dt.platform.ops.service.impl;

import com.alibaba.csp.sentinel.util.StringUtil;
import com.alibaba.fastjson.JSONObject;
import com.dt.platform.constants.enums.ops.MonitorMethodEnum;
import com.dt.platform.domain.ops.MonitorNode;
import com.dt.platform.domain.ops.MonitorTplIndicator;
import com.dt.platform.ops.service.IMonitorAlertService;
import com.dt.platform.ops.service.IMonitorDataProcessPingService;
import com.dt.platform.ops.service.IMonitorTplTriggerService;
import com.github.foxnic.api.error.ErrorDesc;
import com.github.foxnic.api.transter.Result;
import com.github.foxnic.commons.concurrent.SimpleJoinForkTask;
import com.github.foxnic.commons.log.Logger;
import com.github.foxnic.dao.data.Rcd;
import com.github.foxnic.dao.spec.DAO;
import org.github.foxnic.web.framework.dao.DBConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * PingMonitor
 * 支持 Windows / macOS / Linux
 * IPv4 使用 ping，IPv6 使用 ping6
 * 支持每 summaryInterval 次输出一次统计
 */
@Service("MonitorDataProcessPingService")
public class MonitorDataProcessPingServiceImpl  implements IMonitorDataProcessPingService {

    /**
     * 注入DAO对象
     * */
    @Resource(name= DBConfigs.PRIMARY_DAO)
    private DAO dao=null;



    /**
     * 获得 DAO 对象
     * */
    public DAO dao() { return dao; }

    @Autowired
    private MonitorDataProcessBaseServiceImpl monitorDataProcessBaseService;

    @Autowired
    private IMonitorAlertService monitorAlertService;

    @Autowired
    private IMonitorTplTriggerService monitorTplTriggerService;


   // ExecutorService executor = Executors.newFixedThreadPool(3000);

    @Override
    public Result collectData() {

        //后期可以继续优化
        List<MonitorNode> nodeList=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.PING.code());
        Logger.info("collectData,find node number:"+nodeList.size());
        // 创建 ForkJoin 工具，其中 输入一个 Integer 元素的 List ，输出 Long 元素的 List ，每个线程处理 若干元素 ，此处为 5 个
        if(nodeList.size()>0) {
            int batchSize=monitorDataProcessBaseService.calBathSizeValue(nodeList.size());
            Logger.info("current batch size:"+batchSize);
            SimpleJoinForkTask<MonitorNode, Result> task = new SimpleJoinForkTask<>(nodeList, batchSize);
            // 并行执行
            List<Result> rvs = task.execute(els -> {
                // 打印当前线程信息
                Logger.info(Thread.currentThread().getName());
                // 处理当前分到的 若干元素，此处为 5 个
                List<Result> rs = new ArrayList<>();
                for (MonitorNode node : els) {
                    Result result = collectNodeData(node, null, false);
                    if(result.isSuccess()){
                        Logger.debug("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }else{
                        Logger.error("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }
                    rs.add(result);
                }
                // 处理完毕，返回本批次的处理结果，由 SimpleJoinForkTask 负责合并全部结果，并由 rvs 变量接收
                return rs;
            });
        }
        return ErrorDesc.success();
    }

    @Override
    public Result collectNodeData(MonitorNode node, String indicatorId, boolean isForce) {
        List<MonitorTplIndicator> monitorTplIndicatorList=monitorDataProcessBaseService.queryExecuteIndicatorList(node.getId(),"ping",null,isForce);
        queryIndicators(monitorTplIndicatorList,node);
        return ErrorDesc.success();
    }
    public void queryIndicators(List<MonitorTplIndicator> indicators,MonitorNode node){
        if(indicators!=null){
            Logger.info("collectData indicator,find node:"+node.getNodeNameShow()+",indicator number:"+indicators.size());
            for(int i=0;i<indicators.size();i++){
                boolean insRes=queryIndicatorData(indicators.get(i),node);
            }
        }else{
           Logger.info("collectData indicator,find node:"+node.getNodeNameShow()+",indicator null");
        }

    }
    public boolean queryIndicatorData(MonitorTplIndicator indicator,MonitorNode node){
        if(!indicator.getStatus().equals("enable")){
             return false;
        }
        String interval="10";
        String totalCount="60";
        String summaryInterval="10";
        String isFailedByContinuousCount="2";
        String type="";
        String rIp="";
        if(indicator.getCommand().equals("pingV4")){
            type="pingV4";
            rIp=node.getNodeIp();
        }else if(indicator.getCommand().equals("pingV6")){
            type="pingV6";
            rIp=node.getNodeIpV6();
        }else{
            return false;
        }
        if(StringUtil.isBlank(rIp)){
            Logger.error("当前节点ip配置错误,ip:"+rIp,"node:"+node.getNodeNameShow()+",indicator_code:"+indicator.getCode());
            return false;
        }
        //pingV4,{interval:5,totalCount:10,summaryInterval:10}
        String varJsonStr=indicator.getIndicatorVariable();
        if(!StringUtil.isBlank(varJsonStr)){
            JSONObject pingV4V6Json= JSONObject.parseObject(varJsonStr);
            if(pingV4V6Json!=null){
                if (pingV4V6Json.containsKey("isFailedByContinuousCount")){
                    isFailedByContinuousCount=pingV4V6Json.getString("isFailedByContinuousCount");
                }
               if (pingV4V6Json.containsKey("interval")){
                   interval=pingV4V6Json.getString("interval");
                }
                if (pingV4V6Json.containsKey("totalCount")){
                    totalCount=pingV4V6Json.getString("totalCount");
                }
                if (pingV4V6Json.containsKey("summaryInterval")){
                    summaryInterval=pingV4V6Json.getString("summaryInterval");
                }
            }
        }
        String nodeVarJsonStr=node.getVar();
        if(!StringUtil.isBlank(nodeVarJsonStr)){
            JSONObject nodeVarJson=JSONObject.parseObject(nodeVarJsonStr);
            if(nodeVarJson!=null){
               if(nodeVarJson.containsKey(type)){
                   JSONObject pingV4V6Json=nodeVarJson.getJSONObject(type);
                   if(pingV4V6Json!=null){
                       if (pingV4V6Json.containsKey("isFailedByContinuousCount")){
                           isFailedByContinuousCount=pingV4V6Json.getString("isFailedByContinuousCount");
                       }
                       if (pingV4V6Json.containsKey("interval")){
                           interval=pingV4V6Json.getString("interval");
                       }
                       if (pingV4V6Json.containsKey("totalCount")){
                           totalCount=pingV4V6Json.getString("totalCount");
                       }
                       if (pingV4V6Json.containsKey("summaryInterval")){
                           summaryInterval=pingV4V6Json.getString("summaryInterval");
                       }
                   }
               }
            }
        }
        //调用频率要<interval*totalCount
        //一般定时任务外面调用频率30秒
        // interval:建议10s
        // totalCount:建议60个
        // summaryInterval建议10个
        int repeat=Integer.valueOf(interval)*2;
        String ckSql="select 1 from ops_monitor_node_value_last where indicator_code=? and record_time>NOW() - INTERVAL "+repeat+" SECOND and node_id=?";
        dao.pausePrintThreadSQL();
        Rcd rs=dao.queryRecord(ckSql,indicator.getCode(),node.getId());
        dao.resumePrintThreadSQL();
        if(rs!=null){
            Logger.info("ping check,ping maybe is running,node:"+node.getNodeNameShow());
            return true;
        }else{
            Logger.info("ping check,ping is not running,node:"+node.getNodeNameShow()+",call,code:"+indicator.getCode()+",interval:"+Integer.valueOf(interval)+",summaryInterval:"+Integer.valueOf(summaryInterval)+",totalCount:"+totalCount);
        }
        MonitorDataProcessPingUtil pm = new MonitorDataProcessPingUtil();
        if(Integer.valueOf(isFailedByContinuousCount)>0){
             pm.setIsFailedByContinuousCount(Integer.valueOf(isFailedByContinuousCount));
        }
        ExecutorService executor = Executors.newSingleThreadExecutor();
        pm.setNode(node);
        pm.setDao(dao);
        pm.setIndicator(indicator);
        pm.setIpAddress(rIp);
        pm.setPingType(type);
        pm.setIntervalSeconds(Integer.valueOf(interval));
        pm.setSummaryInterval(Integer.valueOf(summaryInterval));
        pm.setTotalCount(Integer.valueOf(totalCount));
        pm.setMonitorTplTriggerService(monitorTplTriggerService);
        pm.setMonitorAlertService(monitorAlertService);
        MonitorDataProcessPingExecutor.addTask(pm);


//        executor.shutdown();
//        try {
//            if (!executor.awaitTermination(1000, TimeUnit.SECONDS)) {
//                Logger.info("node:"+node.getNodeNameShow()+",code:"+indicator.getCode()+",try executor.shutdownNow()");
//                executor.shutdownNow(); // 超时后尝试立即关闭
//            }
//        } catch (InterruptedException e) {
//            executor.shutdownNow(); // 处理中断异常后立即关闭
//            Logger.info("node:"+node.getNodeNameShow()+",code:"+indicator.getCode()+",重置中断状态");
//            Thread.currentThread().interrupt(); // 重置中断状态
//        }
        return true;
    }

}


